﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;
using System.Data.SqlClient;
using AsyncParameters = System.Tuple<System.Action<QueueUnlimited.API.GenericEvent>, string, string>;
using System.Threading;
using System.Diagnostics;
using System.Runtime.Serialization;
using System.IO;
namespace QueueUnlimited.API
{
    public class MessageListener : IDisposable
    {
        private Task task;
        private CancellationTokenSource cancellation;
        public event EventHandler<MessageProcessingFaultEventArgs> OnMessageProcessingFault;

        public MessageListener(Action<GenericEvent> messageProcessor, string connectionString, Subscriber subscriber)
        {
            if (messageProcessor == null) throw new ArgumentNullException("messageProcessor");
            if (string.IsNullOrEmpty(connectionString)) throw new ArgumentNullException("connectionString");

            cancellation = new CancellationTokenSource();
            this.task = new Task((state) => ListenAndProcess(state as AsyncParameters), new AsyncParameters(messageProcessor, connectionString, "SQ_" + subscriber.Name), cancellation.Token);
            this.task.Start();
        }

        public void ListenAndProcess(AsyncParameters parameters)
        {
            while (true && !cancellation.IsCancellationRequested)
            {
                using (var scope = new TransactionScope())
                {
                    GenericEvent nextMessage = null;
                    try
                    {
                        nextMessage = WaitBlockingForNextMessageOrNull(parameters.Item2, parameters.Item3);
                        if (!cancellation.IsCancellationRequested)
                        {
                            if (nextMessage != default(GenericEvent))
                            {
                                parameters.Item1(nextMessage);
                            }
                            scope.Complete();
                        }
                    }
                    catch (Exception ex)
                    {
                        var faultEvent = OnMessageProcessingFault;
                        if (faultEvent != null)
                            faultEvent(this, new MessageProcessingFaultEventArgs(ex, nextMessage));
                    }
                }
            }
        }

        private GenericEvent WaitBlockingForNextMessageOrNull(string connectionString, string queueName)
        {
            using (var c = new SqlConnection(connectionString))
            {
                using (var command = new SqlCommand("WaitForAndReceiveNextMessage", c))
                {
                    command.CommandType = System.Data.CommandType.StoredProcedure;
                    command.Parameters.AddWithValue("@QueueName", queueName);
                    c.Open();
                    var reader = command.ExecuteReader(System.Data.CommandBehavior.SingleResult | System.Data.CommandBehavior.SingleRow | System.Data.CommandBehavior.CloseConnection);
                    if (reader.Read())
                    {
                        return GenericEvent.From(reader.GetString(0));
                    }
                    else
                    {
                        return null;
                    }
                }
            }
        }


        public void Dispose()
        {
            if (task != null && !task.IsCanceled)
            {
                cancellation.Cancel();
            }
            if (task != null)
            {
                task.Wait();
                task.Dispose();
            }
        }
    }

    public class MessageProcessingFaultEventArgs : EventArgs
    {
        public Exception Exception { get; private set; }
        public GenericEvent CurrentMessage { get; private set; }

        internal MessageProcessingFaultEventArgs(Exception ex, GenericEvent current)
        {
            Exception = ex;
            CurrentMessage = current;
        }
    }
}
